package com.gtp.mq.demo.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

import com.gtp.mq.demo.RocketConstants;

public abstract class BaseProducer {

	protected  DefaultMQProducer producer;
	
	public BaseProducer() {
		
		//线程安全的
		
		/*
		 * #producerGroup
		 * 生产者组从概念上汇总了角色完全相同的所有生产者实例
		 * 尤其是在涉及交易消息时很重要
		 * 对于非事务性消息，只要它在每个进程中都是唯一的，就没有关系
		 * 
		 * #这些都是干嘛的? 好像可以收集消息轨迹,发送到内部topic
		 * final String namespace, 
		 * RPCHook rpcHook,
         * boolean enableMsgTrace, 
         * final String customizedTraceTopic
		 */
		producer = new DefaultMQProducer("Producer1");
		producer.setSendMsgTimeout(3000);
		producer.setCompressMsgBodyOverHowmuch( 1024 * 4);//4kb
		
		/*发送失败之前在内部执行的最大重试次数,这可能会导致消息重复，这取决于应用程序开发人员来解决*/
		producer.setRetryTimesWhenSendFailed(2);
		producer.setRetryTimesWhenSendAsyncFailed(2);
		
		//是否在内部发送失败时重试另一个代理
		producer.setRetryAnotherBrokerWhenNotStoreOK(false);
		
		//最大消息:FIXME 最好和服务端一致
		producer.setMaxMessageSize(1024 * 1024 * 4);//4m
		
		/*
		 *  客户端的一些配置可参考:ClientConfig
		    private String namesrvAddr = NameServerAddressUtils.getNameServerAddresses();
		    private String clientIP = RemotingUtil.getLocalAddress();
		    private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");
		   
		    private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();
		    protected String namespace;
		    protected AccessChannel accessChannel = AccessChannel.LOCAL;
		    
		    private int pollNameServerInterval = 1000 * 30;  //定期从NameSrv拉取topic路由信息
		    private int heartbeatBrokerInterval = 1000 * 30; //和broker的心跳(有什么用?感知broker是否挂了?从NameServer你拉取了啊)
		    private int persistConsumerOffsetInterval = 1000 * 5; //持久化Consumer消费进度间隔时间
		    private long pullTimeDelayMillsWhenException = 1000; //发生异常时,延后多少再pull
		    
		    private boolean unitMode = false;
		    private String unitName;
		    private boolean vipChannelEnabled = 
		    private boolean useTLS = TlsSystemConfig.tlsEnable;
		    private LanguageCode language = LanguageCode.JAVA;
		 */
		 //只能设置一个域名,不支持带分隔符的多个地址
		 producer.setNamesrvAddr(RocketConstants.NAMESRV_ADDR);
		//producer.setClientIP("");//RemotingUtil.getLocalAddress()
		//producer.setInstanceName("");//System.getProperty("rocketmq.client.name", "DEFAULT")
        //producer.setPollNameServerInterval(1000 * 30);//从name server提取数据间隔
        //异步回调线程池大小,默认取cpu核数
        //producer.setClientCallbackExecutorThreads(Runtime.getRuntime().availableProcessors());
        //producer.setHeartbeatBrokerInterval(30000);//向Broker发送心跳间隔时间，单位毫秒
        //producer.setPersistConsumerOffsetInterval(5000);//持久化Consumer消费进度间隔时间，单位毫秒
        
        try {
			producer.start();
		} catch (MQClientException e) {
			e.printStackTrace();
		}
    }
	
	public void close(){
		if(producer!=null){
			 producer.shutdown();
		}
	}
}
